package springkafka.controller;
import cn.hutool.json.JSONUtil;
import springkafka.config.SpringBootKafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/kafka")
@Slf4j
public class KafkaProducerController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostMapping(value = "/produce", produces = "application/json")
    public String produce(@RequestBody Object obj) {
        try {
            String obj2String = JSONUtil.toJsonStr(obj);
            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(SpringBootKafkaConfig.TOPIC_TEST, obj2String);
            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    log.info("Message sent to Kafka successfully: {}", result.toString());
                }

                @Override
                public void onFailure(Throwable ex) {
                    log.error("Failed to send message to Kafka: {}", ex.getMessage(), ex);
                }
            });
            return "success";
        } catch (Exception e) {
            log.error("Error occurred while sending message to Kafka: {}", e.getMessage(), e);
            return "failure";
        }
    }
}


